package com.yj.handler;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.annotation.JsonNaming;
import com.yj.common.RpcRequest;
import com.yj.common.RpcResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.Data;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

@Component
@ChannelHandler.Sharable
//public class RpcClientHandler extends SimpleChannelInboundHandler<String> implements Callable<RpcResponse> {
public class RpcClientHandler extends SimpleChannelInboundHandler<String> {

    private ChannelHandlerContext ctx;  // 保存 ChannelHandlerContext 的实例变量
    RpcRequest rpcRequest;
    RpcResponse rpcResponse;

    private static final Map<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();

    private static final List<ChannelHandlerContext> list = new CopyOnWriteArrayList<>();

    public void setRequest(RpcRequest rpcRequest) {
        this.rpcRequest = rpcRequest;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        RpcResponse response = JSON.parseObject(msg, RpcResponse.class);
        String requestId = response.getRequestId();
        CompletableFuture<RpcResponse> future = pendingRequests.remove(requestId);
        if (future != null) {
            future.complete(response);
        }
    }


    //private void setResponse(String msg) {
    //    this.rpcResponse = JSONUtil.toBean(msg, RpcResponse.class);
    //    // 唤醒等待线程
    //    notify();
    //}

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;  // 保存 ChannelHandlerContext
        list.add(ctx);
    }


    //public synchronized RpcResponse sendMsg(RpcRequest rpcRequest) {
    //    String msg = JSONUtil.toJsonStr(rpcRequest);
    //    this.ctx.writeAndFlush(msg);
    //    try {
    //        wait();
    //    } catch (InterruptedException e) {
    //        throw new RuntimeException(e);
    //    }
    //    return rpcResponse;
    //}

    public CompletableFuture<RpcResponse> sendRequest(RpcRequest request) {
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        pendingRequests.put(request.getRequestId(), future);
        String msg = JSON.toJSONString(request);
        list.get(0).writeAndFlush(msg);
        //ctx.writeAndFlush(msg);
        return future;
    }
}


//@Override
//public CompletableFuture<RpcResponse> call() throws Exception {
//    return sendRequest(rpcRequest);
//}
//}
